package com.myCompany.edu.realtime.sink;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.myCompany.edu.realtime.bean.TableProcess;
import com.myCompany.edu.realtime.util.DruidDSUtil;
import com.myCompany.edu.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author
 * @date 2022/9/17 9:31
 */

//自定义sink
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private DruidDataSource dataSource;
    private Jedis redisClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        //从连接池中获取连接
        dataSource = DruidDSUtil.getDataSource();  //获取连接器

        //获取redis客户端
//        redisClient = RedisUtil.getRedisClient();
    }

    @Override
    public void close() throws Exception {
        if (dataSource != null) {
            dataSource.close();  //关闭连接池
        }
//        if (redisClient != null) {
//            redisClient.close();
//        }
    }

    //流中每进来一条元素执行一次

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value,  //流中的元素
                       Context context) throws Exception {
        JSONObject data = value.f0;//业务数据
        TableProcess tp = value.f1;//配置表数据

        //1.写出到phoenix
        writePhoenix(data, tp);

        //2. 新增一个功能，当维度发生更新的时候，删除缓存
//        delCache(data, tp);
    }

    private void delCache(JSONObject data, TableProcess tp) {
        //当维度发生更新的时候，删除缓存
        String op = data.getString("op");
        if ("update".equals(op)) {
            //当维度更新的时候，删除缓存
            String key = tp.getSinkTable() + ":" + data.getString("id");
            redisClient.del(key);
        }
    }

    private void writePhoenix(JSONObject data, TableProcess tp) throws SQLException {
        //TODO
        // upsert into t(a, b, c) values(?, ?, ?)
        StringBuilder sql = new StringBuilder();
        sql
                .append("upsert into ")
                .append(tp.getSinkTable())
                .append("(")
                //拼接字段
                .append(tp.getSinkColumns())
                .append(") values(")
                //拼接占位符  id, tm_name
                .append(tp.getSinkColumns().replaceAll("[^,]+", "?"))
                .append(")");

        //1. 从连接池获取连接对象
        DruidPooledConnection conn = dataSource.getConnection();

        //2. 通过连接对象得到一个预处理语句
        PreparedStatement ps = conn.prepareStatement(sql.toString());

        //3. 对sql中的占位符赋值
        // upsert into t(a, b, c) values(?, ?, ?)
        String[] columns = tp.getSinkColumns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String columnName = columns[i];
            //需要做非空判断：如果是null，就写入null，否则就写入自己
            ps.setString(i + 1, data.get(columnName) == null ?
                    null : data.get(columnName).toString());
        }

        //4. 执行预处理语句
        ps.execute();

        //5. 提交
        conn.commit();

        //6. 关闭预处理语句
        ps.close();

        //7. 归还连接

        conn.close();  //如果这个连接是自己手动获取的，则是关闭连接，如果是从连接池，则是归还连接给连接池
    }

}
